CrewAI में CapSolver एकीकरण के साथ कैप्चा हल करें

pip install crewai
pip install 'crewai[tools]'
pip install requests

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"

class CaptchaSolverInput(BaseModel):
    """CaptchaSolver टूल के लिए इनपुट स्कीमा।"""
    website_url: str = Field(..., description="CAPTCHA वाले वेबसाइट का URL")
    website_key: str = Field(..., description="CAPTCHA के साइट की चाबी")
    captcha_type: str = Field(default="ReCaptchaV2TaskProxyLess", description="हल करने वाले CAPTCHA के प्रकार")

class CaptchaSolverTool(BaseTool):
    name: str = "captcha_solver"
    description: str = "CapSolver API के उपयोग से CAPTCHA चुनौतियां हल करें। reCAPTCHA v2, v3, Turnstile आदि का समर्थन करता है।"
    args_schema: Type[BaseModel] = CaptchaSolverInput

    def _run(self, website_url: str, website_key: str, captcha_type: str = "ReCaptchaV2TaskProxyLess") -> str:
        # कार्य बनाएं
        create_task_url = "https://api.capsolver.com/createTask"

        task_payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": captcha_type,
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }

        response = requests.post(create_task_url, json=task_payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"कार्य बनाने में त्रुटि: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # परिणाम के लिए पूछताछ
        get_result_url = "https://api.capsolver.com/getTaskResult"

        for _ in range(60):  # अधिकतम 60 प्रयास
            time.sleep(2)

            result_payload = {
                "clientKey": CAPSOLVER_API_KEY,
                "taskId": task_id
            }

            response = requests.post(get_result_url, json=result_payload)
            result = response.json()

            if result.get("status") == "ready":
                solution = result.get("solution", {})
                return solution.get("gRecaptchaResponse") or solution.get("token")
            elif result.get("status") == "failed":
                return f"कार्य विफल: {result.get('errorDescription')}"
        return "CAPTCHA समाधान के लिए अवधि समाप्त हो गई"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class ReCaptchaV2Input(BaseModel):
    """reCAPTCHA v2 सॉल्वर के लिए इनपुट स्कीमा।"""
    website_url: str = Field(..., description="reCAPTCHA v2 वाले वेबसाइट का URL")
    website_key: str = Field(..., description="reCAPTCHA के साइट की चाबी")


class ReCaptchaV2Tool(BaseTool):
    name: str = "recaptcha_v2_solver"
    description: str = "CapSolver के उपयोग से reCAPTCHA v2 चुनौतियां हल करें"
    args_schema: Type[BaseModel] = ReCaptchaV2Input

    def _run(self, website_url: str, website_key: str) -> str:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "ReCaptchaV2TaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }
        return self._solve_captcha(payload)

    def _solve_captcha(self, payload: dict) -> str:
        # कार्य बनाएं
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"त्रुटि: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # परिणाम के लिए पूछताछ
        for attempt in range(60):
            time.sleep(2)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return result["solution"]["gRecaptchaResponse"]
            if result.get("status") == "failed":
                return f"विफल: {result.get('errorDescription')}"

        return "समाधान के लिए अवधि समाप्त हो गई"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class ReCaptchaV3Input(BaseModel):
    """reCAPTCHA v3 सॉल्वर के लिए इनपुट स्कीमा।"""
    website_url: str = Field(..., description="reCAPTCHA v3 वाले वेबसाइट का URL")
    website_key: str = Field(..., description="reCAPTCHA के साइट की चाबी")
    page_action: str = Field(default="submit", description="reCAPTCHA v3 के लिए कार्य पैरामीटर")


class ReCaptchaV3Tool(BaseTool):
    name: str = "recaptcha_v3_solver"
    description: str = "स्कोर-आधारित सत्यापन के साथ reCAPTCHA v3 चुनौतियां हल करें"
    args_schema: Type[BaseModel] = ReCaptchaV3Input

    def _run(
        self,
        website_url: str,
        website_key: str,
        page_action: str = "submit"
    ) -> str:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "ReCaptchaV3TaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key,
                "pageAction": page_action
            }
        }

        # कार्य बनाएं
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"त्रुटि: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # परिणाम के लिए पूछताछ
        for attempt in range(60):
            time.sleep(2)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return result["solution"]["gRecaptchaResponse"]
            if result.get("status") == "failed":
                return f"विफल: {result.get('errorDescription')}"

        return "समाधान के लिए अवधि समाप्त हो गई"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class TurnstileInput(BaseModel):
    """Turnstile सॉल्वर के लिए इनपुट स्कीमा।"""
    website_url: str = Field(..., description="Turnstile वाले वेबसाइट का URL")
    website_key: str = Field(..., description="Turnstile विजेट के साइट की चाबी")


class TurnstileTool(BaseTool):
    name: str = "turnstile_solver"
    description: str = "Cloudflare Turnstile चुनौतियां हल करें"
    args_schema: Type[BaseModel] = TurnstileInput

    def _run(self, website_url: str, website_key: str) -> str:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "AntiTurnstileTaskProxyLess",
                "websiteURL": website_url,
                "websiteKey": website_key
            }
        }

        # कार्य बनाएं
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"त्रुटि: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # परिणाम के लिए पूछताछ
        for attempt in range(60):
            time.sleep(2)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return result["solution"]["token"]
            if result.get("status") == "failed":
                return f"विफल: {result.get('errorDescription')}"

        return "समाधान के लिए अवधि समाप्त हो गई"

import requests
import time
from crewai.tools import BaseTool
from typing import Type
from pydantic import BaseModel, Field

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


class CloudflareChallengeInput(BaseModel):
    """Cloudflare चुनौती सॉल्वर के लिए इनपुट स्कीमा।"""
    website_url: str = Field(..., description="सुरक्षित पृष्ठ का URL")
    proxy: str = Field(..., description="प्रॉक्सी के रूप में: http://user:pass@ip:port")


class CloudflareChallengeTool(BaseTool):
    name: str = "cloudflare_challenge_solver"
    description: str = "Cloudflare 5-सेकंड चुनौती पृष्ठ हल करें"
    args_schema: Type[BaseModel] = CloudflareChallengeInput

    def _run(self, website_url: str, proxy: str) -> dict:
        payload = {
            "clientKey": CAPSOLVER_API_KEY,
            "task": {
                "type": "AntiCloudflareTask",
                "websiteURL": website_url,
                "proxy": proxy
            }
        }

        # कार्य बनाएं
        response = requests.post("https://api.capsolver.com/createTask", json=payload)
        result = response.json()

        if result.get("errorId") != 0:
            return f"त्रुटि: {result.get('errorDescription')}"

        task_id = result.get("taskId")

        # परिणाम के लिए पूछताछ
        for attempt in range(60):
            time.sleep(3)
            result = requests.post(
                "https://api.capsolver.com/getTaskResult",
                json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
            ).json()

            if result.get("status") == "ready":
                return {
                    "cookies": result["solution"]["cookies"],
                    "user_agent": result["solution"]["userAgent"]
                }
            if result.get("status") == "failed":
                return f"विफल: {result.get('errorDescription')}"

        return "समाधान के लिए अवधि समाप्त हो गई"

from selenium import webdriver
from selenium.webdriver.common.by import By

def submit_recaptcha_token(driver, token: str):
    """reCAPTCHA टोकन डालें और फॉर्म जमा करें"""
    # छिपे टेक्स्ट एरिया दृश्यमान बनाएं और टोकन सेट करें
    recaptcha_response = driver.find_element(By.ID, "g-recaptcha-response")
    driver.execute_script("arguments[0].style.display = 'block';", recaptcha_response)
    recaptcha_response.clear()
    recaptcha_response.send_keys(token)

    # फॉर्म जमा करें
    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

def submit_turnstile_token(driver, token: str):
    """Turnstile टोकन डालें और जमा करें"""
    # छिपे इनपुट में टोकन सेट करें
    turnstile_input = driver.find_element(By.NAME, "cf-turnstile-response")
    driver.execute_script("arguments[0].value = arguments[1];", turnstile_input, token)

    # फॉर्म जमा करें
    form = driver.find_element(By.TAG_NAME, "form")
    form.submit()

# उदाहरण: कुकीज़ और यूजर-एजेंट का उपयोग अनुरोध में करें
headers = {
    "User-Agent": "user_agent_value",
    "Cookie": "cookies_value"
}
response = requests.get("https://example.com", headers=headers)

# CapSolver समाधान से कुकीज सेट करें
for cookie in cf_solution["cookies"]:
    session.cookies.set(cookie["name"], cookie["value"])

# समाधान के लिए उपयोग किया गया उपयोक्ता एजेंट सेट करें
headers = {
    "User-Agent": cf_solution["user_agent"]
}

# अब आप सुरक्षित पृष्ठ तक पहुंच सकते हैं
response = session.get(url, headers=headers)
return response.text

### पूर्ण डेटा निकालने का उदाहरण

```python
import requests
import time
from selenium import webdriver
from selenium.webdriver.common.by import By

CAPSOLVER_API_KEY = "YOUR_CAPSOLVER_API_KEY"


def solve_recaptcha(website_url: str, website_key: str) -> str:
    """CapSolver से reCAPTCHA टोकन प्राप्त करें"""
    payload = {
        "clientKey": CAPSOLVER_API_KEY,
        "task": {
            "type": "ReCaptchaV2TaskProxyLess",
            "websiteURL": website_url,
            "websiteKey": website_key
        }
    }

    response = requests.post("https://api.capsolver.com/createTask", json=payload)
    result = response.json()

    if result.get("errorId") != 0:
        raise Exception(f"त्रुटि: {result.get('errorDescription')}")

    task_id = result.get("taskId")

    for _ in range(60):
        time.sleep(2)
        result = requests.post(
            "https://api.capsolver.com/getTaskResult",
            json={"clientKey": CAPSOLVER_API_KEY, "taskId": task_id}
        ).json()

        if result.get("status") == "ready":
            return result["solution"]["gRecaptchaResponse"]
        if result.get("status") == "failed":
            raise Exception(f"असफल: {result.get('errorDescription')}")

    raise Exception("समय सीमा समाप्त हो गई")


def scrape_with_recaptcha(target_url: str, site_key: str):
    """पूरा प्रवाह: reCAPTCHA हल करें → जमा करें → डेटा निकालें"""
    driver = webdriver.Chrome()
    driver.get(target_url)

    try:
        # 1. CAPTCHA हल करें
        token = solve_recaptcha(target_url, site_key)

        # 2. टोकन डालें
        recaptcha_response = driver.find_element(By.ID, "g-recaptcha-response")
        driver.execute_script("arguments[0].style.display = 'block';", recaptcha_response)
        recaptcha_response.clear()
        recaptcha_response.send_keys(token)

        # 3. फॉर्म जमा करें
        driver.find_element(By.TAG_NAME, "form").submit()

        # 4. डेटा निकालें
        time.sleep(3)  # पृष्ठ लोड होने के लिए प्रतीक्षा करें
        return driver.page_source

    finally:
        driver.quit()

from selenium import webdriver
from selenium.webdriver.chrome.options import Options

# CapSolver एक्सटेंशन के साथ Chrome विन्यास करें
chrome_options = Options()
chrome_options.add_extension("path/to/capsolver-extension.crx")

driver = webdriver.Chrome(options=chrome_options)

# एक्सटेंशन ऑटोमैटिक रूप से CAPTCHA को हल करेगा
driver.get("https://example.com/protected-page")

def solve_with_retry(self, payload: dict, max_retries: int = 3) -> str:
    for attempt in range(max_retries):
        try:
            result = self._solve_captcha(payload)
            if result:
                return result
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)  # एक्स्पोनेंशियल बैकऑफ

def check_balance() -> float:
    response = requests.post(
        "https://api.capsolver.com/getBalance",
        json={"clientKey": CAPSOLVER_API_KEY}
    )
    return response.json().get("balance", 0)

from functools import lru_cache
from datetime import datetime, timedelta

captcha_cache = {}

def get_cached_token(website_url: str, website_key: str) -> str:
    cache_key = f"{website_url}:{website_key}"

    if cache_key in captcha_cache:
        token, timestamp = captcha_cache[cache_key]
        if datetime.now() - timestamp < timedelta(minutes=2):
            return token

    # नए CAPTCHA को हल करें
    new_token = solve_captcha(website_url, website_key)
    captcha_cache[cache_key] = (new_token, datetime.now())
    return new_token